/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.register.metadata;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.pingan.haofang.framework.common.functional.lang.data.Try;
import com.pingan.haofang.searchcloud.api.matedata.FieldMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolder;
import com.pingan.haofang.searchcloud.api.matedata.MetaCacheSupport;
import com.pingan.haofang.searchcloud.common.exception.SystemException;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.common.cloud.SolrZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.sql.SQLException;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @author huhailiang937
 * @date 2017年6月16日下午5:01:09
 * @since 1.0.0
 */
@Service("zkIndexMetadataRegister")
@ManagedResource
@ConditionalOnProperty(name = "seachcloud.meta.zk-host")
public class ZkIndexMetadataRegister  implements IndexMetadataHolder {

    protected static SolrZkClient zkClient;
    
    protected static Logger LOG = LoggerFactory.getLogger(ZkIndexMetadataRegister.class);

    /**
     * 失效时间：写入之后的 15分钟失效<br/>
     */
    protected final static Cache<String, IndexMeta> META_CACHE = CacheBuilder.newBuilder()
            .expireAfterWrite(15, TimeUnit.MINUTES).build();

    // JDK ServiceLoader 初始化需无参构造器
    public ZkIndexMetadataRegister() {
        LOG.info("-----> JDK ServiceLoader  IndexMetadataHolder for ZkIndexMetadataRegister init sucess");
    }

    @Autowired
    public ZkIndexMetadataRegister(ApplicationContext applicationContext) {
        String metaZkHost = applicationContext.getEnvironment().getProperty(MetaCacheSupport.ZK_CONFIG_KEY);
        LOG.info("----> Route ZK Register Host : "+ metaZkHost);
        if (null == metaZkHost) {
            throw new RuntimeException("Property [seachcloud.meta.zk-host] is null");
        }
        
        CloudSolrClient.Builder builder = new CloudSolrClient.Builder();
        builder.withZkHost(metaZkHost);
        CloudSolrClient solrClient = builder.build();
        solrClient.connect();
        zkClient = solrClient.getZkStateReader().getZkClient();
        
        MetadataChangeListener listener = new MetadataChangeListener(metaZkHost);
        listener.start(META_CACHE);
    }

    @Override
    public Optional<FieldMeta> findFiledMeta(String indexName, String fieldName) {
        Objects.requireNonNull(indexName, "indexName is null");
        Objects.requireNonNull(fieldName, "fieldName is null");
        return Optional.ofNullable(findIndexMeta(indexName).getFiledMetas().get(fieldName));
    }

    @Override
    public Optional<String> findPKFieldName(String indexName) {
        Objects.requireNonNull(indexName, "indexName is null");

        Map<String, FieldMeta> filedMetaMap = Try.of(() -> findIndexMeta(indexName).getFiledMetas())
                .orElseThrow(RuntimeException::new);

        for (Map.Entry<String, FieldMeta> keyValue : filedMetaMap.entrySet()) {
            String key = keyValue.getKey();
            FieldMeta fieldMeta = keyValue.getValue();
            if (fieldMeta.isPk()) {
                return Optional.of(key);
            }
        }

        return Optional.empty();
    }

    @Override
    public IndexMeta findIndexMeta(String indexName) {
        Objects.requireNonNull(indexName, "indexName is null");
        try {
            IndexMeta meta = META_CACHE.getIfPresent(indexName);
            if (null != meta) {
                return meta;
            }

            String zkMetaPath = MetaCacheSupport.getZKMetaPath(indexName);
            try {
                byte[] datas = zkClient.getData(zkMetaPath, null, null, false);
                IndexMeta metaInZk = MetaCacheSupport.decode(datas, IndexMeta.class);
                META_CACHE.put(indexName, metaInZk);
                return metaInZk;
            } catch (Exception e) {
                LOG.error(String.format("ZkIndexMetadataRegister discover(%s) has error: ", indexName), e);
            }
            return null;
        } catch (Exception e) {
            throw new SystemException("search.indexname.notexist", new String[] { indexName });
        }
    }

    @Override
    public boolean register(IndexMeta indexMeta) {
        Assert.notNull(indexMeta, "indexMeta can not be null");
        Assert.notNull(indexMeta.getIndexName(), "indexMeta can not be null");

        String indexMetaPath = MetaCacheSupport.getZKMetaPath(indexMeta.getIndexName());

        boolean initSucess = initZKPath(indexMetaPath);
        if (!initSucess) {
            return false;
        }

        try {
            String json = new String(MetaCacheSupport.encode(indexMeta));
            zkClient.setData(indexMetaPath, json.getBytes(), false);
            META_CACHE.put(indexMeta.getIndexName(), indexMeta);
            LOG.info(String.format("ZkIndexMetadataRegister register Meta : %s", json));
            return true;
        } catch (Exception e) {
            LOG.error(String.format("ZkIndexMetadataRegister register fail! setData(%s) has error:", indexMetaPath), e);
        }
        return false;
    }

    private boolean initZKPath(String path) {
        try {
            boolean isExists = zkClient.exists(path, false);
            if (!isExists) {
                zkClient.makePath(path, true, false);
            }
            return true;
        } catch (Exception e) {
            LOG.error(String.format("zkClient makePath(%s) has error:", path), e);
        }
        return false;
    }


    @ManagedAttribute
    public Map<String, IndexMeta> getConnectionMap() throws SQLException {
        return META_CACHE.asMap();
    }

}
